//
// Created by konglinglong on 2020/9/8.
//

#include <sys/mman.h>
#include <fcntl.h>
#include "../../include/utils/varRingBuffer.h"

#define VRB_ALIGN_SIZE (64)

enum
{
    VAR_RB_DBG_CODE_BUFF_FULL,
    VAR_RB_DBG_CODE_BUFF_EMPTY,
    VAR_RB_DBG_CODE_W_CONFLICT,
    VAR_RB_DBG_CODE_BUFF_ROLL,
    VAR_RB_DBG_CODE_END
};

static const char* gDebugCodeNameTable[] =
{
    MACRO_STRINGIFY(VAR_RB_DBG_CODE_BUFF_FULL),
    MACRO_STRINGIFY(VAR_RB_DBG_CODE_BUFF_EMPTY),
    MACRO_STRINGIFY(VAR_RB_DBG_CODE_W_CONFLICT),
    MACRO_STRINGIFY(VAR_RB_DBG_CODE_BUFF_ROLL),
};
STATIC_ASSERT(ARRAY_SIZE(gDebugCodeNameTable) == VAR_RB_DBG_CODE_END);

typedef struct
{
    uint32_t check; /* 越界检查 */
    uint32_t reqSize; /* 请求大小 */
    uint32_t totalSize; /* 实际分配大小 */
    uint32_t rollFlag; /* 是否是回滚块 */
} S_VarRbElmntHdr;

struct VarRingBuffer
{
    uint32_t type;
    uint32_t isShm;
    char     shmId[128];
    uint32_t shmSize;
    char* buffer; /* ring buffer地址 */
    uint8_t* rwFlag; /* 读写标志 */
    uint32_t count; /* 有效数据个数 */
    uint32_t size; /* 缓存大小，必须为2的n次方 */
    uint32_t sizeMask; /* 缓存大小，为size-1，用于把%size操作等效转换成&sizeMask */
    uint32_t isMp; /* 是否是多生产者 */
    uint32_t wrIdx; /* 写索引 */
    uint32_t rdIdx; /* 读索引 */
    uint32_t dbgCode[VAR_RB_DBG_CODE_END]; /* 调试信息 */
};

/* rb_chunk_elmnt一定要小于等于VRB_ALIGN_SIZE，保证缓冲最后要么至少有一个rb_chunk_elmnt的位置，要么为0 */
STATIC_ASSERT(sizeof(S_VarRbElmntHdr) <= VRB_ALIGN_SIZE);

static int32_t VarRB_Init(S_VarRingBuffer* self, char *dataBuffer, uint32_t dataBuffSize, char *rwFlagBuffer, uint32_t rwFlagBuffSize, uint32_t type)
{
    switch (type)
    {
    case VAR_RB_TYPE_NONE:
    case VAR_RB_TYPE_SPSC:
        self->isMp = 0;
        break;
    case VAR_RB_TYPE_MPSC:
        self->isMp = 1;
        break;
    default:
        return CSP_RC_ER;
    }

    self->type = type;
    self->rwFlag = (uint8_t*)rwFlagBuffer;
    self->buffer = dataBuffer;
    self->count = 0;
    self->size = dataBuffSize;
    self->sizeMask = dataBuffSize - 1;

    memset(self->rwFlag, 0, rwFlagBuffSize);

    MEMORY_BARRIER();

    return CSP_RC_OK;
}

static S_VarRingBuffer* VarRB_LcmCreate(uint32_t size, uint32_t type)
{
    int32_t rc;
    char* bufferBase;
    char* dataBuffer;
    char* rwFlagBuffer;
    uint32_t dataBuffSize;
    uint32_t totalBuffSize;
    uint32_t rwFlagBuffSize;
    uint32_t power2Size;
    S_VarRingBuffer* self;

    power2Size = 1;
    while (power2Size < size)
    {
        power2Size <<= 1;
    }
    dataBuffSize = power2Size;
    rwFlagBuffSize = power2Size / VRB_ALIGN_SIZE;

    totalBuffSize = CALC_ALIGN_SIZE(sizeof(S_VarRingBuffer), VRB_ALIGN_SIZE) + rwFlagBuffSize + dataBuffSize;

    bufferBase = OS_MemCalloc(1, totalBuffSize);
    if (NULL == bufferBase)
    {
        return NULL;
    }
    self = (S_VarRingBuffer*)bufferBase;

    rwFlagBuffer = bufferBase + CALC_ALIGN_SIZE(sizeof(S_VarRingBuffer), VRB_ALIGN_SIZE);
    dataBuffer = rwFlagBuffer + rwFlagBuffSize;

    rc = VarRB_Init(self, dataBuffer, dataBuffSize, rwFlagBuffer, rwFlagBuffSize, type);
    if (CSP_RC_OK != rc)
    {
        OS_MemFree(bufferBase);
        return NULL;
    }
    self->isShm = 0;

    return self;
}

static S_VarRingBuffer* VarRB_ShmCreate(const char* shmId, uint32_t size, uint32_t type)
{
    int32_t rc;
    int shmFd;
    char* bufferBase;
    char* dataBuffer;
    char* rwFlagBuffer;
    uint32_t dataBuffSize;
    uint32_t totalBuffSize;
    uint32_t rwFlagBuffSize;
    uint32_t power2Size;
    S_VarRingBuffer* self;

    power2Size = 1;
    while (power2Size < size)
    {
        power2Size <<= 1;
    }
    dataBuffSize = power2Size;
    rwFlagBuffSize = power2Size / VRB_ALIGN_SIZE;

    totalBuffSize = CALC_ALIGN_SIZE(sizeof(S_VarRingBuffer), VRB_ALIGN_SIZE) + rwFlagBuffSize + dataBuffSize;

    shmFd = shm_open(shmId, O_CREAT | O_RDWR | O_TRUNC, S_IRWXU | S_IRWXG);
    if (shmFd < 0)
    {
        return NULL;
    }
    if (ftruncate(shmFd, totalBuffSize) < 0)
    {
        shm_unlink(shmId);
        return NULL;
    }
    bufferBase = mmap(NULL, totalBuffSize, PROT_READ | PROT_WRITE, MAP_SHARED, shmFd, 0);
    if (bufferBase == (void*)-1)
    {
        shm_unlink(shmId);
        return NULL;
    }

    self = (S_VarRingBuffer*)bufferBase;
    rwFlagBuffer = bufferBase + CALC_ALIGN_SIZE(sizeof(S_VarRingBuffer), VRB_ALIGN_SIZE);
    dataBuffer = rwFlagBuffer + rwFlagBuffSize;

    rc = VarRB_Init(self, dataBuffer, dataBuffSize, rwFlagBuffer, rwFlagBuffSize, type);
    if (CSP_RC_OK != rc)
    {
        munmap(bufferBase, totalBuffSize);
        shm_unlink(shmId);
        return NULL;
    }
    self->isShm = 1;
    self->shmSize = totalBuffSize;
    strncpy(self->shmId, shmId, sizeof(self->shmId));

    return self;
}

S_VarRingBuffer* VarRB_Create(const char* shmId, uint32_t size, uint32_t type)
{
    if (size <= VRB_ALIGN_SIZE)
    {
        return NULL;
    }

    if (shmId)
    {
        return VarRB_ShmCreate(shmId, size, type);
    }
    else
    {
        return VarRB_LcmCreate(size, type);
    }
}

void VarRB_Destroy(S_VarRingBuffer* self)
{
    char shmId[128];

    if (self->isShm)
    {
        strncpy(shmId, self->shmId, sizeof(self->shmId));
        munmap(self, self->shmSize);
        shm_unlink(shmId);
    }
    else
    {
        OS_MemFree(self);
    }
}

void* VarRB_WriteRequest(S_VarRingBuffer* self, uint32_t size)
{
    int32_t ret;
    uint32_t freeSize;
    uint32_t tailSize;
    uint32_t totalSize;
    uint32_t rollFlag;
    uint32_t currRdIdx;
    uint32_t currWrIdx;
    uint32_t currWrOffset;
    S_VarRbElmntHdr* pCurrWrElmnt;

    /* 增加4字节做越界检查 */
    const uint32_t adjust_size = CALC_ALIGN_SIZE(size + sizeof(S_VarRbElmntHdr) + sizeof(uint32_t), VRB_ALIGN_SIZE);

    while (1)
    {
        currWrIdx = self->wrIdx;
        currWrOffset = currWrIdx & self->sizeMask;

        currRdIdx = self->rdIdx;
        freeSize = self->size - currWrIdx + currRdIdx;
        if (unlikely(adjust_size > freeSize))
        {
            CSP_ADD_DBG_CODE(self, VAR_RB_DBG_CODE_BUFF_FULL);
            return NULL;
        }

        /* 写游标到缓冲末尾的长度 */
        tailSize = self->size - currWrOffset;
        if (unlikely(tailSize < adjust_size))
        {
            rollFlag = 1;
            totalSize = tailSize;
        }
        else
        {
            rollFlag = 0;
            totalSize = adjust_size;
        }

        if (self->isMp)
        {
            ret = COMPARE_AND_SWAP(&self->wrIdx, currWrIdx, currWrIdx + totalSize);
            if (unlikely(!ret))
            {
                CSP_ADD_DBG_CODE(self, VAR_RB_DBG_CODE_W_CONFLICT);
                continue;
            }
        }
        else
        {
            self->wrIdx = currWrIdx + totalSize;
        }

        MACRO_ASSERT(self->rwFlag[currWrOffset / VRB_ALIGN_SIZE] == 0);

        pCurrWrElmnt = (S_VarRbElmntHdr*)(self->buffer + currWrOffset);
        pCurrWrElmnt->check = MEM_CHECK_FLAG;
        pCurrWrElmnt->reqSize = size;
        pCurrWrElmnt->totalSize = totalSize;
        pCurrWrElmnt->rollFlag = rollFlag;

        if (unlikely(pCurrWrElmnt->rollFlag))
        {
            MEMORY_BARRIER();
            self->rwFlag[currWrOffset / VRB_ALIGN_SIZE] = 1;
            CSP_ADD_DBG_CODE(self, VAR_RB_DBG_CODE_BUFF_ROLL);
            continue;
        }

        /* 写入检查标志 */
        *((uint32_t*)((char*)pCurrWrElmnt + sizeof(S_VarRbElmntHdr) + size)) = MEM_CHECK_FLAG;
        return (char*)pCurrWrElmnt + sizeof(S_VarRbElmntHdr);
    }
}

int32_t VarRB_WriteCommit(S_VarRingBuffer* self, void* chunk)
{
    S_VarRbElmntHdr* pCurrWrElmnt;

    pCurrWrElmnt = (S_VarRbElmntHdr*)((char*)chunk - sizeof(S_VarRbElmntHdr));
    MACRO_ASSERT(MEM_CHECK_FLAG == pCurrWrElmnt->check);
    MACRO_ASSERT(MEM_CHECK_FLAG == *((uint32_t*)((char*)pCurrWrElmnt + sizeof(S_VarRbElmntHdr) + pCurrWrElmnt->reqSize)));

    if (VAR_RB_TYPE_NONE != self->type)
    {
        FETCH_AND_ADD(&self->count, 1);
        MEMORY_BARRIER();
    }
    else
    {
        self->count++;
    }

    self->rwFlag[((char*)pCurrWrElmnt - self->buffer) / VRB_ALIGN_SIZE] = 1;

    return CSP_RC_OK;
}

void* VarRB_ReadRequest(S_VarRingBuffer* self, uint32_t* size)
{
    uint32_t currRdIdx;
    uint32_t currRdOffset;
    uint8_t rwFlag;
    S_VarRbElmntHdr* pCurrRdElmnt;

    while (1)
    {
        currRdIdx = self->rdIdx;
        currRdOffset = currRdIdx & self->sizeMask;

        rwFlag = self->rwFlag[currRdOffset / VRB_ALIGN_SIZE];
        if (1 != rwFlag)
        {
            CSP_ADD_DBG_CODE(self, VAR_RB_DBG_CODE_BUFF_EMPTY);
            return NULL;
        }

        pCurrRdElmnt = (S_VarRbElmntHdr*)(self->buffer + currRdOffset);
        MACRO_ASSERT(MEM_CHECK_FLAG == pCurrRdElmnt->check);

        if (unlikely(pCurrRdElmnt->rollFlag))
        {
            self->rwFlag[currRdOffset / VRB_ALIGN_SIZE] = 0;
            MEMORY_BARRIER();
            self->rdIdx += pCurrRdElmnt->totalSize;
            continue;
        }

        MACRO_ASSERT(MEM_CHECK_FLAG ==
                     *((uint32_t*)((char*)pCurrRdElmnt + sizeof(S_VarRbElmntHdr) + pCurrRdElmnt->reqSize)));
        *size = pCurrRdElmnt->reqSize;
        return (char*)pCurrRdElmnt + sizeof(S_VarRbElmntHdr);
    }
}

int32_t VarRB_ReadConsume(S_VarRingBuffer* self, void* chunk)
{
    S_VarRbElmntHdr* pCurrRdElmnt;

    pCurrRdElmnt = (S_VarRbElmntHdr*)((char*)chunk - sizeof(S_VarRbElmntHdr));

    MACRO_ASSERT(MEM_CHECK_FLAG == pCurrRdElmnt->check);
    MACRO_ASSERT(MEM_CHECK_FLAG == *((uint32_t*)((char*)pCurrRdElmnt + sizeof(S_VarRbElmntHdr) + pCurrRdElmnt->reqSize)));

    self->rwFlag[((char*)pCurrRdElmnt - self->buffer) / VRB_ALIGN_SIZE] = 0;

    if (VAR_RB_TYPE_NONE != self->type)
    {
        FETCH_AND_SUB(&self->count, 1);
        MEMORY_BARRIER();
    }
    else
    {
        self->count--;
    }

    self->rdIdx += pCurrRdElmnt->totalSize;

    return CSP_RC_OK;
}

int32_t VarRB_Put(S_VarRingBuffer* self, void* data, uint32_t size)
{
    void* chunk;

    chunk = VarRB_WriteRequest(self, size);
    if (unlikely(NULL == chunk))
    {
        return CSP_RC_ER;
    }
    memcpy(chunk, data, size);
    return VarRB_WriteCommit(self, chunk);
}

int32_t VarRB_Get(S_VarRingBuffer* self, void* buffer, uint32_t* size)
{
    void* chunk;
    uint32_t chunkSize;

    chunk = VarRB_ReadRequest(self, &chunkSize);
    if (unlikely(NULL == chunk || *size < chunkSize))
    {
        return CSP_RC_ER;
    }
    *size = chunkSize;
    memcpy(buffer, chunk, chunkSize);
    return VarRB_ReadConsume(self, chunk);
}

uint32_t VarRB_Count(S_VarRingBuffer* self)
{
    return self->count;
}

void VarRB_Dump(S_VarRingBuffer* self, void* buffer, uint32_t size)
{
    uint32_t loop;
    int32_t offset;

    offset = snprintf(buffer, size, "VAR RING BUFFER<type:%u> : \n", self->type);
    if (offset >= size)
    {
        return;
    }
    for (loop = 0; loop < ARRAY_SIZE(gDebugCodeNameTable); loop++)
    {
        offset += snprintf((char*)buffer + offset, size - offset, "%s : %u\n",
            gDebugCodeNameTable[loop], self->dbgCode[loop]);
        if (offset >= size)
        {
            break;
        }
    }
}


